package com.suning.sawp.service.impl.dao;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.annotation.Resource;

import org.apache.commons.lang3.StringUtils;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.mapreduce.MapReduceResults;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service;

import com.suning.framework.dal.client.DalClient;
import com.suning.sawp.dto.bi.BiStoreYetdSaleCountDto;
import com.suning.sawp.dto.bi.BiStoreYetdSaleCountMapReduce;
import com.suning.sawp.exception.SawpException;
import com.suning.sawp.intf.dao.BiKafkaRealTimeSaleDaoService;
import com.suning.sawp.po.bi.BiKafkaRealTimeSale;
import com.suning.sawp.service.util.DJStringUtils;
import com.suning.sawp.service.util.DateUtils;

@Service("biKafkaRealTimeSaleDaoService")
public class BiKafkaRealTimeSaleDaoServiceImpl implements BiKafkaRealTimeSaleDaoService {

    @Resource
    MongoOperations mongoTemplate;

    @Resource
    DalClient dalClient;

    @Override
    public void upsertBiKafkaRealTimeSale(List<BiKafkaRealTimeSale> kafkaRealTimeSales) {
        mongoTemplate.insert(kafkaRealTimeSales, getTableName());
    }

    /**
     * 
     * 功能描述: 根据当前所在的时刻路由表<br>
     * 〈功能详细描述〉
     * 
     * @return
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    private static String getTableName() {
        // 取当前时间的小时数和分钟数
        Calendar cal = Calendar.getInstance();
        cal.setTime(new Date());
        int minute = cal.get(Calendar.MINUTE);
        int index = 0;
        if (minute >= 30) {
            index = 1;
        }
        return "BiKafkaRealTimeSale_" + index;
    }

    @Override
    public List<BiStoreYetdSaleCountDto> groupCountStoreYetdSaleData() {
        String mapFunc = "function(){emit(this.storeCode, {saleAmount:this.saleAmount, saleVolume:this.saleVolume, storeCode:this.storeCode});}";
        String reduceFunc = "function(key, value){" + "var result = {saleAmount:0, saleVolume:0, storeCode:''};"
                + "for (var i = 0; i < value.length; i++) {" + "result.saleAmount += parseInt(value[i].saleAmount);"
                + "result.saleVolume += value[i].saleVolume;" + "result.storeCode = value[i].storeCode;" + "}"
                + "return result;" + "}";

        return getGroupCountData(mapFunc, reduceFunc, "");
    }

    @Override
    public List<BiStoreYetdSaleCountDto> groupCountStoreCateYetdSaleData(String storeCode) {
        String mapFunc = "function(){emit(this.storeCode+this.categoryCode, {saleAmount:this.saleAmount, saleVolume:this.saleVolume, storeCode:this.storeCode, categoryCode:this.categoryCode});}";
        String reduceFunc = "function(key, value){"
                + "var result = {saleAmount:0, saleVolume:0, storeCode:'', cateCode:''};"
                + "for (var i = 0; i < value.length; i++) {" + "result.saleAmount += parseInt(value[i].saleAmount);"
                + "result.saleVolume += value[i].saleVolume;" + "result.storeCode = value[i].storeCode;"
                + "result.cateCode = value[i].categoryCode;" + "}" + "return result;" + "}";

        return getGroupCountData(mapFunc, reduceFunc, storeCode);
    }

    @Override
    public List<BiStoreYetdSaleCountDto> groupCountStaffYetdSaleData(String storeCode) {
        String mapFunc = "function(){emit(this.staffId, {saleAmount:this.saleAmount, saleVolume:this.saleVolume, staffId:this.staffId, storeCode:this.storeCode});}";
        String reduceFunc = "function(key, value){"
                + "var result = {saleAmount:0, saleVolume:0, staffId:'', storeCode:''};"
                + "for (var i = 0; i < value.length; i++) {" + "result.saleAmount += parseInt(value[i].saleAmount);"
                + "result.saleVolume += value[i].saleVolume;" + "result.staffId = value[i].staffId;"
                + "result.storeCode = value[i].storeCode;" + "}" + "return result;" + "}";

        return getGroupCountData(mapFunc, reduceFunc, storeCode);
    }

    private List<BiStoreYetdSaleCountDto> getGroupCountData(String mapFunc, String reduceFunc, String storeCode) {
        String collectionName = "BiKafkaRealTimeSale_1";
        long totalSize = mongoTemplate.count(new Query(), "BiKafkaRealTimeSale_1");
        if (totalSize == 0) {
            totalSize = mongoTemplate.count(new Query(), "BiKafkaRealTimeSale_0");
            if (totalSize == 0) {
                // 0号表也无数据，抛出异常
                throw new SawpException("接收数据失败");
            } else {
                // 0号表有数据则取0好表数据，门店11点之后不在销售，可以认为11:30的数据等于00:00数据
                collectionName = "BiKafkaRealTimeSale_0";
            }
        }

        List<BiStoreYetdSaleCountDto> list = new ArrayList<BiStoreYetdSaleCountDto>();
        Query query = new Query();
        if (StringUtils.isNotBlank(storeCode)) {
            Criteria criteria = new Criteria();
            criteria.and("storeCode").is(storeCode);
            query.addCriteria(criteria);
        }
        MapReduceResults<BiStoreYetdSaleCountMapReduce> mapReduce = mongoTemplate.mapReduce(query, collectionName,
                mapFunc, reduceFunc, BiStoreYetdSaleCountMapReduce.class);
        if (mapReduce != null) {
            for (BiStoreYetdSaleCountMapReduce count : mapReduce) {
                list.add(count.getValue());
            }
        }

        return list;
    }

    @Override
    public void removeBiKafkaRealTimeSale() {
        // 考虑到删除几十万数据较慢，直接采用drop表的方式，有点简单粗暴
        mongoTemplate.dropCollection("BiKafkaRealTimeSale_0");
        mongoTemplate.dropCollection("BiKafkaRealTimeSale_1");
    }

    @Override
    public BigDecimal queryLastYearStoreSaleAmount(String storeCode, String time) {
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("storeCode", storeCode);
        paramMap.put("time", time);
        paramMap.put("statisDate", DateUtils.getDateStrByFormat(new Date(), DateUtils.YYYYMMDD_PATTERN));
        BigDecimal saleAmount = dalClient.queryForObject("BI_STORE_SALE_REAL_TIME.QUERY_LAST_YEAR_STORE_SALE_AMOUNT",
                paramMap, BigDecimal.class);
        return saleAmount;
    }
    
    @Override
    public Map<String, Object> queryLastYearStoreSaleData(String storeCode, String time) {
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("storeCode", storeCode);
        paramMap.put("time", time);
        paramMap.put("statisDate", DateUtils.getDateStrByFormat(new Date(), DateUtils.YYYYMMDD_PATTERN));
        Map<String, Object> saleMap = dalClient.queryForMap("BI_STORE_SALE_REAL_TIME.QUERY_LAST_YEAR_STORE_SALE_DATA",
                paramMap);
        return saleMap;
    }
	
	    
    @Override
    public  Map<String, Object> queryLastYearStoreSaleAmountEx(String storeCode, String time) {
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("storeCode", storeCode);
        paramMap.put("time", time);
        paramMap.put("statisDate", DateUtils.getDateStrByFormat(new Date(), DateUtils.YYYYMMDD_PATTERN));
        return  dalClient.queryForMap("BI_STORE_SALE_REAL_TIME.QUERY_LAST_YEAR_STORE_SALE_JXC_AMOUNT",
                paramMap);
    }
    
    /**
     * 查询去年同期客单价
     * 功能描述: <br>
     * 〈功能详细描述〉
     *
     * @param storeCode
     * @param time
     * @return
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    @Override
    public BigDecimal queryLastYearStoreCustAvgPrice(String storeCode, String time) {
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("storeCode", storeCode);
        paramMap.put("time", time);
        paramMap.put("statisDate", DateUtils.getDateStrByFormat(new Date(), DateUtils.YYYYMMDD_PATTERN));
        Map<String, Object> sale = dalClient.queryForMap("BI_STORE_SALE_REAL_TIME.QUERY_LAST_YEAR_STORE_CUST_AVG_PRICE",
                paramMap);
        // 数据不为空时,处理客单价
        if (null != sale && !sale.isEmpty() && null != sale.get("amount") && null != sale.get("num")) {
            return DJStringUtils.divide(DJStringUtils.bigDecimal(String.valueOf(sale.get("amount"))), DJStringUtils.bigDecimal(String.valueOf(sale.get("num"))));
        }
        return null;
    }

    @Override
    public BigDecimal queryLastYearStoreCateSaleAmount(String storeCode, String cateCode, String time) {
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("storeCode", storeCode);
        paramMap.put("cateCode", cateCode);
        paramMap.put("time", time);
        paramMap.put("statisDate", DateUtils.getDateStrByFormat(new Date(), DateUtils.YYYYMMDD_PATTERN));
        BigDecimal saleAmount = dalClient.queryForObject(
                "BI_STORE_SALE_REAL_TIME.QUERY_LAST_YEAR_STORE_CATE_SALE_AMOUNT", paramMap, BigDecimal.class);
        return saleAmount;
    }
    
    
    @Override
    public Map<String, Object> queryLastYearStoreCateSaleAmountEx(String storeCode, String cateCode, String time) {
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("storeCode", storeCode);
        paramMap.put("cateCode", cateCode);
        paramMap.put("time", time);
        paramMap.put("statisDate", DateUtils.getDateStrByFormat(new Date(), DateUtils.YYYYMMDD_PATTERN));
        Map<String, Object> result = dalClient.queryForMap(
                "BI_STORE_SALE_REAL_TIME.QUERY_LAST_YEAR_STORE_CATE_SALE_JXC_AMOUNT", paramMap);
        return result;
    }
    

    @Override
    public void batchUpdateStoreRealTimeSale(List<Map<String, Object>> saleDays) {
        List<Map<String, Object>> mapList = new ArrayList<Map<String, Object>>();
        for (Map<String, Object> saleDay : saleDays) {
            Map<String, Object> paramMap = new HashMap<String, Object>();
            paramMap.put("storeCode", saleDay.get("strIdGj"));
            // 总共
            paramMap.put("saleAmount", saleDay.get("payAmt"));
            paramMap.put("saleVolume", saleDay.get("saleCnt"));
            // 正向
            paramMap.put("positiveSaleAmount", saleDay.get("payAmt_f"));
            paramMap.put("positiveSaleVolume", saleDay.get("saleCnt_f"));
            // 逆向
            paramMap.put("negativeSaleAmount", saleDay.get("payAmt_r"));
            paramMap.put("negativeSaleVolume", saleDay.get("saleCnt_r"));
            // 客单
            paramMap.put("custNum", saleDay.get("memberCount"));
            paramMap.put("updateTime", saleDay.get("updateTime"));

            mapList.add(paramMap);
        }

        @SuppressWarnings("unchecked")
        Map<String, Object>[] batchValues = (Map<String, Object>[]) mapList.toArray(new Map[mapList.size()]);
        // 执行SQL操作
        dalClient.batchUpdate("BI_STORE_SALE_REAL_TIME.UPSERT_STORE_REAL_TIME_SALE", batchValues);
    }

    @Override
    public void batchUpdateStoreCateRealTimeSale(List<Map<String, Object>> cateSaleDays) {
        List<Map<String, Object>> mapList = new ArrayList<Map<String, Object>>();
        for (Map<String, Object> saleDay : cateSaleDays) {
            Map<String, Object> paramMap = new HashMap<String, Object>();
            paramMap.put("storeCode", saleDay.get("strIdGj"));
            paramMap.put("cateCode", saleDay.get("deptCd"));
            // 总共
            paramMap.put("saleAmount", saleDay.get("payAmt"));
            paramMap.put("saleVolume", saleDay.get("saleCnt"));
            // 正向
            paramMap.put("positiveSaleAmount", saleDay.get("payAmt_f"));
            paramMap.put("positiveSaleVolume", saleDay.get("saleCnt_f"));
            // 逆向
            paramMap.put("negativeSaleAmount", saleDay.get("payAmt_r"));
            paramMap.put("negativeSaleVolume", saleDay.get("saleCnt_r"));
            // 客单
            paramMap.put("custNum", saleDay.get("memberCount"));
            paramMap.put("updateTime", saleDay.get("updateTime"));
            mapList.add(paramMap);
        }

        @SuppressWarnings("unchecked")
        Map<String, Object>[] batchValues = (Map<String, Object>[]) mapList.toArray(new Map[mapList.size()]);
        // 执行SQL操作
        dalClient.batchUpdate("BI_STORE_SALE_REAL_TIME.UPSERT_STORE_CATE_REAL_TIME_SALE", batchValues);
    }

    @Override
    public BigDecimal querySaleAmountByStoreCd(String storeCode) {
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("storeCode", storeCode);
        return dalClient.queryForObject("BI_STORE_SALE_REAL_TIME.QUERY_STORE_REAL_TIME_SALE", paramMap,
                BigDecimal.class);
    }

    @Override
    public Map<String, Object> querySaleAmountByStoreCd4Alias(String storeCode) {
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("storeCode", storeCode);
        return dalClient.queryForMap("BI_STORE_SALE_REAL_TIME.QUERY_STORE_REAL_TIME_SALE_4_ALIAS", paramMap);
    }

    @Override
    public BigDecimal querySaleAmountByStoreCdCateCd(String storeCode, String cateCode) {
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("storeCode", storeCode);
        paramMap.put("cateCode", cateCode);
        return dalClient.queryForObject("BI_STORE_SALE_REAL_TIME.QUERY_STORE_CATE_REAL_TIME_SALE", paramMap,
                BigDecimal.class);
    }
    
    @Override
    public Map<String, Object> querySaleAmountByStoreCdCateCd4Alias(String storeCode, String cateCode) {
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("storeCode", storeCode);
        paramMap.put("cateCode", cateCode);
        return dalClient.queryForMap("BI_STORE_SALE_REAL_TIME.QUERY_STORE_CATE_REAL_TIME_SALE_4_ALIAS", paramMap);
    }

    /**
     * 查询门店所有品类的实时数据
     */
    @Override
    public List<Map<String, Object>> queryCatesSaleAmountByStoreCdAlias(String storeCode) {
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("storeCode", storeCode);
        return dalClient.queryForList("BI_STORE_SALE_REAL_TIME.QUERY_STORE_REAL_TIME_SALE_4_ALL_CATE", paramMap);
    }

    /**
     * 查询门店品类同期销量
     */
    @Override
    public BigDecimal queryLastYearStoreCateSaleVolume(String storeCode, String cateCode, String time) {
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("storeCode", storeCode);
        paramMap.put("cateCode", cateCode);
        paramMap.put("time", time);
        paramMap.put("statisDate", DateUtils.getDateStrByFormat(new Date(), DateUtils.YYYYMMDD_PATTERN));
        return dalClient.queryForObject(
                "BI_STORE_SALE_REAL_TIME.QUERY_LAST_YEAR_STORE_CATE_SALE_VOLUME", paramMap, BigDecimal.class);
    }

    @Override
    public Map<String, Object> queryLastYearSaleByStoreCdCateCd(String storeCode, String cateCode, String statisTime) {
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("storeCode", storeCode);
        paramMap.put("cateCode", cateCode);
        paramMap.put("time", statisTime);
        paramMap.put("statisDate", DateUtils.getDateStrByFormat(new Date(), DateUtils.YYYYMMDD_PATTERN));
        return dalClient.queryForMap("BI_STORE_SALE_REAL_TIME.QUERY_LAST_YEAR_STORE_CATE_SALE", paramMap);
    }

    @Override
    public List<Map<String, Object>> queryLastYearStoreSaleCateList(String storeCode, String time) {
        Map<String, Object> paramMap = new HashMap<String, Object>();
        paramMap.put("storeCode", storeCode);
        paramMap.put("time", time);
        paramMap.put("statisDate", DateUtils.getDateStrByFormat(new Date(), DateUtils.YYYYMMDD_PATTERN));
        return dalClient.queryForList("BI_STORE_SALE_REAL_TIME.QUERY_LAST_YEAR_STORE_CATE_SALE_LIST", paramMap);
    }

}
